package org.apache.ocean;

import java.io.File;
import java.sql.PreparedStatement;
import java.sql.Types;
import java.util.List;
import java.util.logging.Logger;

import org.apache.commons.lang.SerializationUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.lucene.document.Document;
import org.apache.ocean.database.JDBCDatabase;
import org.apache.ocean.database.SQLDefaultParameters;
import org.apache.ocean.database.SQLParameter;
import org.apache.ocean.database.SQLParameters;
import org.apache.ocean.database.SQLResults;
import org.apache.ocean.database.SQLRow;

/**
 * 
 * @author jasonr
 */
public class SQLTransactionDatabase extends JDBCDatabase {
	public static Logger log = Logger.getLogger(SQLTransactionDatabase.class.getName());
	public TransactionRecordTransform transactionRecordTransform = new TransactionRecordTransform();
	public static TransactionDefaults TRANSACTION_DEFAULTS = new TransactionDefaults();

	static {
		System.setProperty("derby.storage.pageSize", "32768");
	}

	public SQLTransactionDatabase(File directory) throws Exception {
		super("transactionRecord", directory);
		createTable("transactionRecord", TransactionRecord.class);
		syncColumns("transactionRecord");
	}
  
	public CLong findMaxVersionByDocumentID(CLong documentID) throws Exception {
		String sql = "SELECT MAX(version) FROM transactionRecord WHERE documentID = ?";
		SQLResults results = executeQuery(sql, new SQLParameters(documentID), "transactionRecord");
		return results.getAggregateLong();
	}
	
	public TransactionRecord findVersionInfoByDocumentID(CLong documentID) throws Exception {
		String sql = "SELECT id, documentID, version, dateCreated FROM transactionRecord WHERE documentID = ? ORDER BY version DESC";
		SQLResults results = executeQuery(sql, new SQLParameters(documentID, Types.BIGINT), "transactionRecord", new CLong(1));
		return (TransactionRecord)results.getFirst(transactionRecordTransform);
	}
	
	public void deleteTransactions(List<TransactionID> transactionIDs) throws Exception {
		String sql = "DELETE FROM transactionRecord " + "WHERE id IN (" + ClusterUtil.concat("?", ", ", transactionIDs.size()) + ")";
		SQLParameters parameters = new SQLParameters(transactionIDs.size());
		for (TransactionID transactionID : transactionIDs) {
			parameters.add(new SQLParameter(transactionID, Types.DECIMAL));
		}
		executeUpdate(sql, parameters, "transactionRecord");
	}

	public int findTransactionRecordCountByStatus(Integer status) throws Exception {
		String sql = "SELECT COUNT(id) FROM transactionRecord WHERE status = ?";
		SQLResults results = executeQuery(sql, new SQLParameters(status, Types.INTEGER), "transactionRecord");
		return results.getCount();
	}

	public int findTransactionRecordCount() throws Exception {
		String sql = "SELECT COUNT(id) FROM transactionRecord";
		SQLResults results = executeQuery(sql, null, "transactionRecord");
		return results.getCount();
	}

	public TransactionID findTransactionIDHighestReal() throws Exception {
		String sql = "SELECT MAX(id) FROM transactionRecord";
		SQLResults results = executeQuery(sql, null, "transactionRecord");
		return getTransactionID(results);
	}

	public static TransactionID getTransactionID(SQLResults results) {
		SQLRow row = results.getFirstRow();
		if (row != null) {
			Object object = row.getValue(0);
			if (object != null) {
				return new TransactionID(object.toString());
			}
		}
		return null;
	}
  
	public TransactionID findCellTransactionIDHighest() throws Exception {
		String sql = "SELECT MAX(cellTransactionID) FROM transactionRecord";
		SQLResults results = executeQuery(sql, null, "transactionRecord");
		return getTransactionID(results);
	}
	
	public TransactionID findTransactionIDHighest() throws Exception {
		String sql = "SELECT MAX(id) FROM transactionRecord";
		SQLResults results = executeQuery(sql, null, "transactionRecord");
		return getTransactionID(results);
	}

	public TransactionRecord findLastTransactionRecord() throws Exception {
		TransactionID id = findTransactionIDHighest();
		return findTransactionRecord(id);
	}

	public TransactionRecord findTransactionRecord(TransactionID id) throws Exception {
		return findTransactionRecord(null, id);
	}

	public class TransactionRecordTransform implements SQLResults.Transform<TransactionRecord> {
		public TransactionRecord row(SQLRow row) throws Exception {
			return toTransactionRecord(row);
		}
	}

	public TransactionRecord findTransactionRecord(List<String> columnNames, TransactionID id) throws Exception {
		String columnString = createColumnsString(columnNames);
		String sql = "SELECT " + columnString + " FROM transactionRecord WHERE id = ?";
		SQLResults results = executeQuery(sql, new SQLParameters(id, Types.NUMERIC), "transactionRecord");
		return (TransactionRecord) results.getFirst(transactionRecordTransform);
	}

	public TransactionRecord toTransactionRecord(SQLRow row) throws Exception {
		return (TransactionRecord) rowToObject(row, TransactionRecord.class, TRANSACTION_DEFAULTS);
	}

	public void updateTransactionRecordStatus(TransactionRecord transactionRecord) throws Exception {
		TransactionRecord updateSet = new TransactionRecord();
		updateSet.id = transactionRecord.id;
		updateSet.status = transactionRecord.status;
		updateSet.message = transactionRecord.message;
		updateSet.stackTrace = transactionRecord.stackTrace;
		updateSet.dateModified = transactionRecord.dateModified;
		updateSet.thisStatus = transactionRecord.thisStatus;
		updateSet.thisMessage = transactionRecord.thisMessage;
		updateSet.thisStackTrace = transactionRecord.thisStackTrace;
		updateSet("transactionRecord", updateSet, false, TRANSACTION_DEFAULTS);
	}

	public Results findTransactionRecordsResults(List<String> columnNames) throws Exception {
		return findTransactionRecordsResults(columnNames, "ORDER BY id ASC");
	}

	public Results findTransactionRecordsResults(List<String> columnNames, String orderBy) throws Exception {
		String columnString = createColumnsString(columnNames);
		String sql = "SELECT " + columnString + " FROM transactionRecord " + orderBy;
		return executeQueryResults(sql, null, "transactionRecord", TransactionRecord.class, TRANSACTION_DEFAULTS);
	}

	public Results findTransactionRecordResults(TransactionID id) throws Exception {
		return findTransactionRecordGreaterResults(id, null);
	}

	public List<TransactionRecord> findTransactionRecords(TransactionID id, CLong rows) throws Exception {
		return findTransactionRecordsGreater(id, null, rows);
	}

	public List<TransactionRecord> findTransactionRecordsGreater(TransactionID id, List<String> columnNames, CLong rows) throws Exception {
		String columnString = "*";
		if (columnNames != null) {
			columnString = StringUtils.join(columnNames.iterator(), ", ");
		}
		String sql = "SELECT " + columnString + " FROM transactionRecord " + "WHERE id >= ? ORDER BY id ASC";
		SQLParameters parameters = new SQLParameters(id, Types.DECIMAL);
		SQLResults results = executeQuery(sql, parameters, "transactionRecord", rows);
		return results.toList(transactionRecordTransform);
	}

	public static class TransactionDefaults extends SQLDefaultParameters<TransactionRecord> {
		public boolean isSetObject(Object object) throws Exception {
			return (object instanceof Document);
		}

		public void setObject(int index, Object object, PreparedStatement preparedStatement) throws Exception {
			if (object instanceof Document) {
				Document document = (Document) object;
				byte[] bytes = SerializationUtils.serialize(document);
				preparedStatement.setBytes(index, bytes);
			}
		}

		public boolean isGetObject(String name) throws Exception {
			return StringUtils.equals("document", name);
		}

		public Object getObject(String name, Object value) throws Exception {
			if (StringUtils.equals("document", name)) {
				byte[] bytes = (byte[]) value;
				return SerializationUtils.deserialize(bytes);
			}
			return null;
		}

		public void setValues(String action, TransactionRecord transactionRecord) {
			if (StringUtils.equals("update", action)) {
				transactionRecord.dateModified = new java.util.Date();
			} else if (StringUtils.equals("insert", action)) {
				transactionRecord.dateModified = new java.util.Date();
				transactionRecord.dateCreated = transactionRecord.dateModified;
			}
		}
	}

	public Results findTransactionRecordLessResults(TransactionID id, List<String> columnNames) throws Exception {
		String sql = "SELECT " + createColumnsString(columnNames) + " FROM transactionRecord " + "WHERE id < ? ORDER BY id ASC";
		SQLParameters parameters = new SQLParameters(id, Types.DECIMAL);
		return executeQueryResults(sql, parameters, "transactionRecord", TransactionRecord.class, TRANSACTION_DEFAULTS);
	}

	public Results findTransactionRecordGreaterResults(TransactionID id, List<String> columnNames) throws Exception {
		String sql = "SELECT " + createColumnsString(columnNames) + " FROM transactionRecord " + "WHERE id >= ? ORDER BY id ASC";
		SQLParameters parameters = new SQLParameters(id, Types.DECIMAL);
		return executeQueryResults(sql, parameters, "transactionRecord", TransactionRecord.class, TRANSACTION_DEFAULTS);
	}

	public void updateTransactionRecord(TransactionRecord transactionRecord) throws Exception {
		updateRow("transactionRecord", transactionRecord, TRANSACTION_DEFAULTS);
	}

	public void insertTransactionRecords(List<TransactionRecord> transactionRecords) throws Exception {
		insertBatch("transactionRecord", transactionRecords, TRANSACTION_DEFAULTS);
	}

	public TransactionRecord insertTransactionRecord(TransactionRecord transactionRecord) throws Exception {
		return (TransactionRecord) insert("transactionRecord", transactionRecord, TRANSACTION_DEFAULTS);
	}
}